package cn.erbudev;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.BiFunction;

/*
 * 
 * xd_zhiwang 
 */
public class CommunicableThread extends Thread {
	private long id;
	private ConcurrentHashMap<Long, PairStream> connectedMap;
	private Queue<DistributedTask> tasks; // 任务列表
	private Condition running = new Condition(true);

	public CommunicableThread(long id) {
		this.id = id;
		this.connectedMap = new ConcurrentHashMap<Long, PairStream>();
		this.tasks = new LinkedList<DistributedTask>();
		this.running.setConnectedMap(connectedMap);
	}

	public void run() {
		while (running.isRunning()) {
			try {
				ObjectInputStream inobj = null;
				for (Iterator<Map.Entry<Long, PairStream>> it = connectedMap.entrySet().iterator(); it.hasNext();) {
					Map.Entry<Long, PairStream> entry = it.next();
					if (entry.getValue().isOn) {
						try {
							inobj = new ObjectInputStream(entry.getValue().getIn());
						} catch (IOException e) {
							continue;
						}

						DistributedTask task = (DistributedTask) inobj.readObject();
						tasks.add(task);
					}
				}
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			}

			while (!tasks.isEmpty()) {
				DistributedTask task = tasks.poll();
				if (task.isFinish) {
					for (PairStream p : connectedMap.values()) {
						if (p.isOn)
							p.setClose();
					}
					running.setRunning(false);
				} else {
					PairStream connected = connectedMap.get(task.targetThreadID);
					System.out.println("Thread:" + task.targetThreadID);
					DistributedTask newTask = task.compute(running);

					System.out.println("newThread:" + task.targetThreadID);
					ObjectOutputStream outobj;
					try {
						outobj = new ObjectOutputStream(connected.getOut());
						outobj.writeObject(newTask);
					} catch (IOException e) {
						e.printStackTrace();
					}
				}
			}
		}
	}

	public long getId() {
		return id;
	}

	public CommunicableThread setId(long id) {
		this.id = id;
		return this;
	}

	public void addConnectableThread(CommunicableThread other) {
		if (this.connectedMap.containsKey(other.getId())) {
			return;
		}
		PipedOutputStream pipedOutputStream1 = new PipedOutputStream();
		// 创建管道输入流
		PipedInputStream pipedInputStream1 = new PipedInputStream();
		try {
			// 将管道输入流与输出流连接 此过程也可通过重载的构造函数来实现
			pipedOutputStream1.connect(pipedInputStream1);
		} catch (IOException e) {
			e.printStackTrace();
		}
		PipedOutputStream pipedOutputStream2 = new PipedOutputStream();
		// 创建管道输入流
		PipedInputStream pipedInputStream2 = new PipedInputStream();
		try {
			// 将管道输入流与输出流连接 此过程也可通过重载的构造函数来实现
			pipedOutputStream2.connect(pipedInputStream2);
		} catch (IOException e) {
			e.printStackTrace();
		}

		PairStream p1 = new PairStream(pipedOutputStream1, pipedInputStream2);
		PairStream p2 = new PairStream(pipedOutputStream2, pipedInputStream1);
		p1.setBrother(p2);
		this.connectedMap.compute(other.id, new BiFunction<Long, PairStream, PairStream>() {
			@Override
			public PairStream apply(Long t, PairStream u) {
				return p1;
			}
		});
		other.connectedMap.compute(this.id, new BiFunction<Long, PairStream, PairStream>() {
			@Override
			public PairStream apply(Long t, PairStream u) {
				return p2;
			}
		});
	}

	public ConcurrentHashMap<Long, PairStream> getConnectedMap() {
		return connectedMap;
	}

	static class PairStream {
		private PipedOutputStream out;
		private PipedInputStream in;
		private PairStream brother;
		private volatile boolean isOn = true;

		public PairStream(PipedOutputStream out, PipedInputStream in) {
			this.out = out;
			this.in = in;
		}

		public PipedOutputStream getOut() {
			return out;
		}

		public PipedInputStream getIn() {
			return in;
		}

		public PairStream getBrother() {
			return brother;
		}

		public void setBrother(PairStream brother) {
			this.brother = brother;
			brother.brother = this;
		}

		public synchronized boolean isOn() {
			return isOn;
		}

		public synchronized void setClose() {
			isOn = false;
			brother.isOn = false;

			try {
				in.close();
				out.close();
				brother.out.close();
				brother.in.close();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

		}
	}

	static class Condition {
		private boolean running;
		private ConcurrentHashMap<Long, PairStream> connectedMap;

		public Condition(boolean running) {
			super();
			this.running = running;
		}

		public boolean isRunning() {
			return running;
		}

		public void setRunning(boolean running) {
			this.running = running;
		}

		public ConcurrentHashMap<Long, PairStream> getConnectedMap() {
			return connectedMap;
		}

		public void setConnectedMap(ConcurrentHashMap<Long, PairStream> connectedMap) {
			this.connectedMap = connectedMap;
		}

	}
}
